package com.zhaosc.spark

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object TrackLogAnalysis {

  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("wxlog")
      .setMaster("local");
    val sc = new SparkContext(conf);
    val fileRdd = sc.textFile("data1.txt");
//    fileRdd.filter(_.startsWith("WXATrackLog4Analysis"))
//      .map(c =>
//        {
//          val params = c.substring("WXATrackLog4Analysis".length()).split(",")
//          val p = params(params.length - 1).split(":");
//          //          new Tuple2(p(0), p(1).substring(0, p(1).length() - 1));
//          p(1).substring(0, p(1).length() - 1);
//        }).map((_, 1)).reduceByKey(_ + _).foreach(println _);

    fileRdd.filter(_.startsWith("WXATrackLog4Analysis")).map(c =>
        {
          val params = c.substring("WXATrackLog4Analysis".length()).split(",")
          val p = params(params.length - 1).split(":");
          //          new Tuple2(p(0), p(1).substring(0, p(1).length() - 1));
          p(1).substring(0, p(1).length() - 1);
        }).distinct().map((_, 1)).reduceByKey(_ + _).saveAsTextFile("hdfs://127.0.0.1:9000/user/zhaosc/input/vp");

    sc.stop()

  }
}